package ex.datastream;

import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class IncrementalAggregation extends ApiFrame {
    public static void main(String[] args) throws Exception {
        IncrementalAggregation windows = new IncrementalAggregation();
        windows.getEnv();
        DataStream<Tuple2<String, Long>> input = windows.getsocketTextStreamText();
        input.keyBy(k -> k.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .aggregate(new AverageAggregate());
        input.print();

        windows.env.execute();
    }

    /**
     * a,6
     * a,2
     * a,1
     输入值:6, 计数1 累加值:6
     输入值:2, 计数2 累加值:8
     输入值:1, 计数3 累加值:9
     累加值:9.0 计数:3 平均值:3.0
     2> (a,3.0)
     *
     *select t.key1,sum(value1)/count(value1) from incrementalaggregationtest t group by key1
     */
    private static class AverageAggregate
            implements AggregateFunction<Tuple2<String, Long>, Tuple2<Long, Long>, Double> {
        @Override
        public Tuple2<Long, Long> createAccumulator() {
            return new Tuple2<>(0L, 0L);
        }

        @Override
        public Tuple2<Long, Long> add(Tuple2<String, Long> value, Tuple2<Long, Long> accumulator) {
            long count=accumulator.f1 + 1L;
            long sum=accumulator.f0 + value.f1;
            System.out.println(" 输入值:"+value.f1+", 计数"+count+" 累加值:"+sum);
            return new Tuple2<>(accumulator.f0 + value.f1, accumulator.f1 + 1L);
        }

        @Override
        public Double getResult(Tuple2<Long, Long> accumulator) {
            System.out.println("累加值:"+ ((double) accumulator.f0) +" 计数:"+ accumulator.f1+" 平均值:"+ ((double) accumulator.f0) / accumulator.f1);
            return ((double) accumulator.f0) / accumulator.f1;
        }

        @Override
        public Tuple2<Long, Long> merge(Tuple2<Long, Long> a, Tuple2<Long, Long> b) {
            return new Tuple2<>(a.f0 + b.f0, a.f1 + b.f1);
        }
    }

    private static class MyProcessWindowFunction
            extends ProcessWindowFunction<Double, Tuple2<String, Double>, String, TimeWindow> {

        public void process(String key,
                            Context context,
                            Iterable<Double> averages,
                            Collector<Tuple2<String, Double>> out) {
            Double average = averages.iterator().next();
            out.collect(new Tuple2<>(key, average));
        }
    }
}
